/*
 * 文件名：DataDistributeThread.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：yuanze
 * 修改时间：2014年11月14日
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.thread;

import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.realstream.server.dao.RealStreamUnitDao;
import com.ailk.bigdata.etl.realstream.server.dao.impl.RealStreamUnitDaoPGImpl;
import com.ailk.bigdata.etl.realstream.server.model.RealDataDispatcher;
import com.ailk.bigdata.etl.realstream.server.model.RealStreamDataPack;
import com.ailk.bigdata.etl.realstream.server.model.RealStreamUnit;
import com.ailk.bigdata.etl.realstream.server.tools.RealStreamTool;

/**
 * @description 数据分发线程
 * @author [yuanze]
 * @version [版本号,2014年11月14日]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class DataDistributeThread implements Runnable
{
	private static Logger logger = LoggerFactory.getLogger(DataDistributeThread.class);
	private LinkedBlockingQueue<RealStreamDataPack> QUEUE = null;

	public DataDistributeThread(LinkedBlockingQueue<RealStreamDataPack> QUEUE)
	{
		this.QUEUE = QUEUE;
	}

	@Override
	public void run()
	{
		logger.info("实时流数据分发线程启动............");
		while (true)
		{
			RealStreamDataPack rDataPack = null;
			try
			{
				rDataPack = QUEUE.take();
				RealStreamUnit streamUnit = null;
				long dataType = rDataPack.getDataType();
				if (!RealStreamTool.REALS_UNIT_CACHE.containsKey(dataType))
				{
					RealStreamUnitDao unitDao = new RealStreamUnitDaoPGImpl();
					streamUnit = unitDao.getUnitAllByUnitCode(dataType);
					if (null == streamUnit)
					{
						logger.info("单元接口类型不存在.........");
						continue;
					}
					RealStreamTool.REALS_UNIT_CACHE.put(dataType, streamUnit);
				}
				else
				{
					streamUnit = RealStreamTool.REALS_UNIT_CACHE.get(dataType);
				}
				// 格式化
				// rDataPack.format(streamUnit.getAcceptFormat(),
				// streamUnit.getSendFormat());
				// RealStreamTool.BACK_FILESYS_DATA_QUEUE.put(rDataPack);
				// 待发送队列
				RealDataDispatcher.addMessageQueue(rDataPack, streamUnit);
			}
			catch (InterruptedException e)
			{
				logger.error("data distribute exception ......", e);
			}
			catch (Exception e)
			{
				logger.error("data format exception ......", e);
			}
		}
	}

}
